package com.headfirst.dmp.report.AppAnalyzeRpt

import com.headfirst.dmp.beans.LogBean
import com.headfirst.dmp.utils.{AreaAnalyzeUtils, RedisUtils}
import org.apache.commons.lang.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.JedisCluster

/**
  * 媒体报表分析
  *
  * 使用工具 spark-core
  *
  * 本次假设规则数据随时在变，因此将其放入redis中
  *
  */
object AppAnalyzeRPT_Redis {


  def main(args: Array[String]): Unit = {


    //1.验证参数
    if (args.length != 3) {
      print(
        """
          |com.headfirst.dmp.report.AreaAnalyzeRPT_V2
          |需要参数：
          |       logInputPath
          |       resultOutputPath
          |       rulesFileInputPath
        """.stripMargin)
      sys.exit(-1)
    }

    //2.接受参数
    val Array(logInputPath, resultOutputPath, rulesFileInputPath) = args


    //3.创建saprkcontext对象
    val conf: SparkConf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local[*]") //本地测试使用local，提交到集群则注释掉该配置
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //设置序列化方式采用KryoSerializer方式（默认的是java序列化）

    val context: SparkContext = new SparkContext(conf)


    //4.读取文件并处理
    context.textFile(logInputPath)
      .map(_.split(",", -1))
      .filter(_.length >= 85)
      .map(LogBean(_))
      .filter(t => !t.appid.isEmpty || !t.appname.isEmpty)
      .mapPartitions(iter => {
        val jedis: JedisCluster = RedisUtils.getRedisCluster()
        val tuples: Iterator[((String, String), List[Double])] = iter.map(log => {
          var appname: String = log.appname
          if (StringUtils.isBlank(appname)) {
            appname = jedis.get(log.appid)
          }
          //原始请求、有效请求、广告请求
          val reqList: List[Double] = AreaAnalyzeUtils.caculateReq(log.requestmode, log.processnode)
          //参与竞价的次数，参与竞价的次数
          val bidList: List[Double] = AreaAnalyzeUtils.caculateBid(log.adplatformproviderid, log.iseffective, log.isbilling, log.isbid, log.adorderid, log.iswin)
          //展示数,点击数
          val showList: List[Double] = AreaAnalyzeUtils.caculateShowAndClick(log.requestmode, log.iseffective)
          //广告消费 ，广告成本
          val costList: List[Double] = AreaAnalyzeUtils.caculateCost(log.adplatformproviderid, log.iseffective, log.isbilling, log.iswin, log.adorderid, log.adcreativeid, log.winprice, log.adpayment)

          //返回值
          ((log.provincename, log.cityname), reqList ++ bidList ++ showList ++ costList)

        })

        jedis.close();

        tuples

      }).reduceByKey((list1, list2) => {list1.zip(list2).map(t => t._1 + t._2)})
        .map(t => {t._1._1 + "," + t._1._2 + "," + t._2.mkString(",")})
        .saveAsTextFile(resultOutputPath)

    //关流
    context.stop()


  }

}
